package com.xpc.rpc.consumer.handler;

import com.xpc.rpc.protocol.ProtocolMessage;
import com.xpc.rpc.protocol.request.RpcRequest;
import com.xpc.rpc.protocol.response.RpcResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class RpcConsumerHandler extends SimpleChannelInboundHandler<ProtocolMessage<RpcResponse>> {


    private static final Logger LOGGER = LoggerFactory.getLogger(RpcConsumerHandler.class);

    private Channel channel;

    private Map<Long,Object> handleResultMap = new ConcurrentHashMap<>();

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        this.channel = ctx.channel();
    }


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ProtocolMessage<RpcResponse> protocolMessage) throws Exception {
        handleResultMap.put(protocolMessage.getRpcHeader().getRequestId(),protocolMessage.getT().getData());
    }

    public Object sendRequest(ProtocolMessage<RpcRequest> requestProtocolMessage) {
        channel.writeAndFlush(requestProtocolMessage);
        while(true) {
            Object result = handleResultMap.remove(requestProtocolMessage.getRpcHeader().getRequestId());
            if(result != null) {
                return result;
            }
        }
    }


    public Channel getChannel() {
        return channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }
}
